package com.yanggu.flink.datastream_api.offlinealarm

import cn.hutool.core.date.DateUtil
import cn.hutool.json.JSONUtil
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.util.Date
import scala.beans.BeanProperty

//这种只适用于设备只有一个离线规则的情况, 如果一个设备有多个离线规则就不行了
//数据流和配置流进行connect, 判定设备是否离线
//两个测试数据
//{"deviceId":"deviceId","delete":false,"systemCode":"systemCode","delayTime":15000}
//{"cimCode":"cimCode","deviceId":"deviceId","systemCode":"systemCode","time":1643346340607,"value":0}
object OffAlarmDemo {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    environment.setParallelism(1)

    val dataStream = environment
      //设备测点数据流
      .socketTextStream("hadoop100", 8000)
      .map(data => JSONUtil.toBean(data, classOf[CimPoint]))
      //根据设备分组
      .keyBy(data => (data.systemCode, data.deviceId))

    val configDataStream = environment
      //设备离线规则流
      .socketTextStream("hadoop100", 9000)
      .map(data => JSONUtil.toBean(data, classOf[DeviceConfig]))
      //根据设备分组
      .keyBy(data => (data.systemCode, data.deviceId))

    dataStream
      .connect(configDataStream)
      .process(new KeyedCoProcessFunction[(String, String), CimPoint, DeviceConfig, String] {

        /**
         * 最大离线时间
         */
        private lazy val maxOffLineTimeState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("maxOffLineTimeState", classOf[Long]))

        /**
         * 最新数据的时间戳
         */
        private lazy val dataTimeState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("dataTimeState", classOf[Long]))

        /**
         * 定时器执行时间戳
         */
        private lazy val triggerTimeStampState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("triggerTimeStampState", classOf[Long]))

        /**
         * 是否离线的状态, true表示离线, false表示在线
         * 默认为true表示离线, 用于恢复之前的离线告警
         */
        private lazy val offLineState = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("offLineState", classOf[Boolean], true))

        /**
         * 测点数据流根据是否有离线规则和测点数据的时间是否大于上次的时间, 更新dataTimeState状态
         *
         * @param value
         * @param ctx
         * @param out
         */
        override def processElement1(value: CimPoint,
                                     ctx: KeyedCoProcessFunction[(String, String), CimPoint, DeviceConfig, String]#Context,
                                     out: Collector[String]): Unit = {
          val maxOffLineTime = maxOffLineTimeState.value()
          val triggerTime = triggerTimeStampState.value()
          //判断是否有离线规则, 如果没有直接return
          if (maxOffLineTime == 0 && triggerTime == 0) {
            return
          }
          //如果数据的时间大于上次的时间就更新dataTimeState
          val time = value.time
          if (time > dataTimeState.value()) {
            dataTimeState.update(time)
            //如果离线了, 现在设备上传数据了发送数据表示设备在线了, 恢复之前的离线告警
            if (offLineState.value()) {
              offLineState.update(false)
              out.collect(s"设备: ${ctx.getCurrentKey}在线, 重新上传数据时间: ${DateUtil.formatDateTime(new Date(time))}" +
                s", 恢复之前所有的离线告警, 当前时间: ${DateUtil.formatDateTime(new Date())}")
            }
            //删除之前的定时器, 重新注册定时器, 保存定时器执行时间
            ctx.timerService().deleteProcessingTimeTimer(triggerTime)
            val newTriggerTime = System.currentTimeMillis() + maxOffLineTimeState.value()
            ctx.timerService().registerProcessingTimeTimer(newTriggerTime)
            triggerTimeStampState.update(newTriggerTime)
          }
        }

        //配置数据流
        override def processElement2(value: DeviceConfig,
                                     ctx: KeyedCoProcessFunction[(String, String), CimPoint, DeviceConfig, String]#Context,
                                     out: Collector[String]): Unit = {

          val triggerTime = triggerTimeStampState.value()
          val timerService = ctx.timerService()

          //如果是删除离线告警规则, 清空所有状态和销毁定时器
          if (value.delete) {
            maxOffLineTimeState.clear()
            triggerTimeStampState.clear()
            dataTimeState.clear()
            offLineState.clear()
            timerService.deleteProcessingTimeTimer(triggerTime)
            return
          }

          val lastMaxOffLineTime = maxOffLineTimeState.value()
          val currentDelayTime = value.delayTime

          //如果是第一次设置时间, 注册定时器, 更新定时器的执行时间
          if (lastMaxOffLineTime == 0 && triggerTime == 0) {
            val triggerTimeStamp = System.currentTimeMillis() + currentDelayTime
            timerService.registerProcessingTimeTimer(triggerTimeStamp)
            triggerTimeStampState.update(triggerTimeStamp)
            //如果设置时间发生了变化且不是第一次更新时间
          } else if (lastMaxOffLineTime != 0 && triggerTime != 0 && lastMaxOffLineTime != currentDelayTime) {
            //删除原有的定时器
            timerService.deleteProcessingTimeTimer(triggerTime)
            val newTriggerTime = triggerTime - lastMaxOffLineTime + currentDelayTime
            //注册新的定时器
            timerService.registerProcessingTimeTimer(newTriggerTime)
            //更新定时器执行的时间戳
            triggerTimeStampState.update(newTriggerTime)
          }

          //直接更新状态即可
          maxOffLineTimeState.update(currentDelayTime)
        }

        override def onTimer(timestamp: Long, ctx: KeyedCoProcessFunction[(String, String), CimPoint, DeviceConfig, String]#OnTimerContext, out: Collector[String]): Unit = {
          val dataTime = dataTimeState.value()
          val delay = maxOffLineTimeState.value()
          if (dataTime + delay < timestamp) {
            out.collect(s"设备: ${ctx.getCurrentKey}离线, 上一次上传数据的时间: ${DateUtil.formatDateTime(new Date(dataTime))}, " +
              s"离线了${(timestamp - dataTime) / 1000}秒, 离线时间: ${DateUtil.formatDateTime(new Date(timestamp))}")
            //更新状态表示离线了
            offLineState.update(true)
          }

          //继续注册定时器, 检测是否离线
          val timerService = ctx.timerService()
          val triggerTimeStamp = timestamp + delay
          timerService.registerProcessingTimeTimer(triggerTimeStamp)
          triggerTimeStampState.update(triggerTimeStamp)
        }
      })
      .print("OffLineAlarm>>>>>")

    environment.execute("OffAlarmDemo01 Job")

  }

}

case class CimPoint(@BeanProperty var systemCode: String,
                    @BeanProperty var deviceId: String,
                    @BeanProperty var cimCode: String,
                    @BeanProperty var time: Long,
                    @BeanProperty var value: Double)

case class DeviceConfig(@BeanProperty var systemCode: String,
                        @BeanProperty var deviceId: String,
                        @BeanProperty var delayTime: Long,
                        @BeanProperty var delete: Boolean)